package org.nbict.iot.persist.repo;

import org.influxdb.InfluxDB;
import org.influxdb.InfluxDBFactory;
import org.influxdb.dto.Point;
import org.influxdb.dto.Pong;
import org.nbict.iot.PropertyUtil;
import org.nbict.iot.persist.Repo;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Map;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

/**
 * Created by songseven on 18/6/16.
 */
public class InfluxRepo implements Repo {

    private String url;
    private String username;
    private String password;
    private String database;
    private String retention;

    private InfluxDB influxDB;
    private static final Logger log = LoggerFactory.getLogger(InfluxRepo.class);
    private ScheduledExecutorService pingSchedule = Executors
            .newSingleThreadScheduledExecutor();

    public InfluxRepo() {
        init();
    }

    private void init() {

        url = PropertyUtil.getProperty("influx.url");
        username = PropertyUtil.getProperty("influx.username");
        password = PropertyUtil.getProperty("influx.password");
        database = PropertyUtil.getProperty("influx.database");
        retention = PropertyUtil.getProperty("influx.retention.policy");

        influxDB = InfluxDBFactory.connect(url, username, password);
        influxDB.enableBatch(2000, 100, TimeUnit.MILLISECONDS, Executors.defaultThreadFactory(), (failedPoints, throwable) -> {
            log.warn("==> error");
        });

//        String rpName = "aRetentionPolicy";
        influxDB.setDatabase(database);
        influxDB.enableGzip();
//        influxDB.createRetentionPolicy(rpName, database, "30d", "30m", 2, true);
//        influxDB.setRetentionPolicy(rpName);

        //监控是否可用
        pingSchedule.scheduleAtFixedRate(() -> {
            try {
                Pong pong = influxDB.ping();
                log.info("ping influxdb:{}", pong);
            } catch (Exception e) {
                log.error("failed to ping influxdb", e);
            }
        }, 10, 10,
        TimeUnit.SECONDS);
    }

    /**
     * 插入一个点
     * @param measurement
     * @param tags
     * @param fields
     * @param timestamp
     * @return
     */
    public void insert(String measurement, Map tags, Map fields, Long timestamp) {
        Point point = Point.measurement(measurement).time(timestamp, TimeUnit
                .MILLISECONDS).tag(tags).fields(fields).build();
        influxDB.write(point);
    }

    public void insert(Point point) {
        influxDB.write(point);
    }

    public void batchInsert() {

    }

    //TODO influxdb安装，测试，时序数据批量写入，改为配置文件方式
    //TODO storm集群安装

//    public static void main(String[] args) throws InterruptedException {
//
//
//        InfluxRepo repo = (InfluxRepo) DbRepository.INFLUX.getInstance();
////        Map tag = new HashMap();
////        tag.put("vin", "LJU70X1Z6HG552141");
////        for (int i = 0; i < 100; i++) {
////            Map fields = new HashMap();
////            double lat = new Double(-30 + (int) (Math.random() * 75));
////            double lng = new Double(-120 + (int) (Math.random() * 70));
////            long time = System.currentTimeMillis();
////            fields.put("lon", lng);
////            fields.put("lat", lat);
////            repo.insert("car_gis", tag,  fields, System.currentTimeMillis());
////            Thread.sleep(100);
////        }
//        Query query = new Query("SELECT * FROM \"car_gis\"", "ocean");
//        QueryResult queryResult = repo.influxDB.query(query);
//
//        InfluxDBResultMapper resultMapper = new InfluxDBResultMapper();
//        List<DTO> testList = resultMapper.toPOJO(queryResult, DTO.class);
//        testList.forEach(e->{
//            System.out.println(e);
//        });
//    }
}
